import java.net.{URLEncoder, URL}

import org.apache.http.client.HttpClient
import org.apache.http.impl.client.DefaultHttpClient
import org.apache.spark.SparkConf

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.io.Source

/**
  * Created by laiweiwei on 15/11/20.
  */
object InvertedIndex extends Setup {

  def main(args: Array[String]): Unit = {

      //----- 倒排索引 -----
      /*
       *  1. files
       *     dir
       *      |-- file1.txt
       *      |-- file2.txt
       *      |-- fileN.txt
       *  2. content&segment words
       *     file1.txt -> 中华人民共和国的面积辽阔 -> 中华 人民 共和 共和国 面积 辽阔
       *     file2.txt -> 美利坚合众国的面积辽阔 -> 美利坚 面积 辽阔
       *     file3.txt -> 苏维埃联邦的面积巨大 -> 苏维埃 联邦 面积 巨大
       *  3. index
       *     中华 -> file1.txt
       *     美利坚 -> file2.txt
       *     面积 -> file1.txt, file2.txt, file3.txt
       *     辽阔 -> file1.txt, file2.txt
       *     巨大 -> file3.txt
       *     苏维埃 -> file3.txt
       *  4. search
       *     input : 哪个国家面积巨大?
       *     segment : 国家 面积 巨大
       *     match :
       *       面积 -> file1.txt, file2.txt, file3.txt
       *       巨大 -> file3.txt
       *     result: file3.txt
       */
      var path = "hdfs://10.17.1.30:9000/vivi/"
      //val path = "file:///Users/laiweiwei/doc"
      val dir = ctx.wholeTextFiles(path)

      dir
        .filter(f => !f._1.contains("java-vs-scala")) // 不要这篇文章
        .map(f => {
          val list = new ArrayBuffer[(String, String)]()
          // 调用分词API
          val url = "http://5.tbip.sinaapp.com/api.php?str="+URLEncoder.encode(f._2, "UTF-8")+"&type=str"
          val lines = Source.fromURL(url).getLines()
          for (line <- lines) {
            line.split(",").foreach(w => {
              list.append(( w, f._1 ))
            })
          }
          list
        })
        .flatMap(list => list)
        .reduceByKey((v1, v2) => v1+","+v2)
        .collect()
        .foreach(println)

      ctx.stop

  }

}
